from pykafka import KafkaClient

client = KafkaClient(hosts="10.10.205.90:9092")
print(client.topics)
topic = client.topics['mykafka']

#synchronously mode
with topic.get_sync_producer() as producer:
    for i in range(4):
        message = 'test message' + str(i ** 2)
        msg = message.encode("utf-8")
        producer.produce(msg)

#asynchronous mode
with topic.get_sync_producer(delivery_reports=True) as producer:
    count = 0
    while True:
        count += 1
        producer.produce('test msg', partition_key='{}'.format(count))
        if count % 10 ** 5 == 0:
            while True:
                try:
                    msg, exc = producer.get_delivery_report(block=False)
                    if exc is not None:
                        print('Failed to deliver msg {}:{}'.format(msg.partition_key, repr(exc)))
                    else:
                        print('Successfully delivered msg {}'.format(msg.partition_key)
                except Queue.Emty:
                    break

consumer = topic.get_simple_consumer()
for message in consumer:
    if message is not None:
        print(message.offset, message.value)


balanced_consumer = topic.get_balanced_consumer(
    consumer_group='',
    auto_commit_enable=True,
    #zookeeper_connect=''
)